package org.xyf.IM.handler;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.StringUtils;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.TextWebSocketHandler;
import org.springframework.web.util.UriComponentsBuilder;
import org.xyf.IM.pojo.KafkaMsg;
import org.xyf.common.constant.GlobalConstant;

import java.io.IOException;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;


@Slf4j
public class WebsocketHandler extends TextWebSocketHandler {

    public static final Map<String, WebSocketSession> sessionMap = new ConcurrentHashMap<>();


    private ObjectMapper objectMapper;

    private KafkaTemplate<String,String> kafkaTemplate;


    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
        String userId = UriComponentsBuilder.fromUri(session.getUri())
                .build()
                .getQueryParams()
                .getFirst(GlobalConstant.USER_ID_HEADER);// 自动处理
        if (!StringUtils.hasLength(userId)) {
            session.sendMessage(new TextMessage("你咋回事,咋没有userId,请检查token!"));
            session.close();
            return;
        }
        sessionMap.put(userId, session);
    }

    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
        super.afterConnectionClosed(session, status);
    }

    @Override
    protected void handleTextMessage(WebSocketSession session, TextMessage message) throws IOException {

        if (Thread.currentThread().isVirtual()) {
            log.info("当前线程为虚拟线程!");
        }

        log.info("收到消息:" + message.getPayload());

        try {
            KafkaMsg msg = objectMapper.readValue(message.getPayload(), KafkaMsg.class);
            if (Objects.isNull(msg)) {
                return;
            }
            if (StringUtils.hasLength(msg.getTargetTopic()) && Objects.nonNull(msg.getBody())) {
                this.sendToTopic(msg,session);
            }

        } catch (Exception e) {
            log.error("handleTextMessage Error: ", e);
            session.sendMessage(new TextMessage("发生了错误"));
        }

    }


    private void sendToTopic(KafkaMsg message,WebSocketSession session) throws JsonProcessingException {
        String msgBody = objectMapper.writeValueAsString(message.getBody());
        CompletableFuture<SendResult<String, String>> future = kafkaTemplate.send(message.getTargetTopic(), msgBody);
        future.whenComplete( (res,err) -> {
            if(Objects.nonNull(err)){
                log.error("sendToTopic error: ",err);
                try {
                    session.sendMessage(new TextMessage("发送消息异常，请稍后重试!"));
                } catch (IOException e) {
                    log.error("在发送消息时出现了io异常: ",e);
                }
            }
        });
    }


    public WebsocketHandler(ObjectMapper objectMapper,KafkaTemplate<String,String> kafkaTemplate) {
        this.objectMapper = objectMapper;
        this.kafkaTemplate = kafkaTemplate;
    }
}
